package com.shux.kafka.consumer;

import java.nio.ByteBuffer;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteBufferDeserializer;
import org.apache.kafka.common.serialization.BytesDeserializer;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.utils.Bytes;

/**
 * 描述：
 * 
 * <pre>HISTORY
 * ****************************************************************************
 *  ID   DATE           PERSON          REASON
 *  1    2017年9月18日      Simba.Hua         Create
 * ****************************************************************************
 * </pre>
 * @author Simba.Hua
 */
public class KafkaConsumerRegister {
	private static final Map<String,ShuxKafkaConsumer<?>> topicMap = new ConcurrentHashMap<>();
	private static <T> boolean register(ConsumerConfig config,ConsumerOptionConfig consumerOptionConfig,IMessageListener<T> listener,Deserializer<T> deserializer){
		if (!topicMap.containsKey(config.getTopic())) {
			synchronized(topicMap){
				if (!topicMap.containsKey(config.getTopic())) {
					ShuxKafkaConsumer<T> consumer ;
					if( !consumerOptionConfig.getTransation() ) {
						consumer = new TransactionalKafkaConsumer<>(config,consumerOptionConfig,listener,deserializer);
					} else {
						consumer = new NoTransactionalKafkaConsumer<>(config,consumerOptionConfig,listener,deserializer);
					}
					topicMap.put(config.getTopic(), consumer);
					return true;
				}
			}
		}
		return false;
	}
	public static  boolean registerString(ConsumerConfig config,ConsumerOptionConfig consumerOptionConfig,IMessageListener<String> listener){
		return register(config,consumerOptionConfig,listener,new StringDeserializer());
	}
	public static  boolean registerLong(ConsumerConfig config,ConsumerOptionConfig consumerOptionConfig,IMessageListener<Long> listener){
		return register(config,consumerOptionConfig,listener,new LongDeserializer());
	}
	public static boolean registerBytes(ConsumerConfig config,ConsumerOptionConfig consumerOptionConfig,IMessageListener<Bytes> listener){
		return register(config,consumerOptionConfig,listener,new BytesDeserializer());
	}
	
	public static boolean registerByteBuffer(ConsumerConfig config,ConsumerOptionConfig consumerOptionConfig,IMessageListener<ByteBuffer> listener){
		
		return register(config,consumerOptionConfig,listener,new ByteBufferDeserializer());
	}
	
	public static boolean registerByteArray(ConsumerConfig config,ConsumerOptionConfig consumerOptionConfig,IMessageListener<byte[]> listener){
		
		return register(config,consumerOptionConfig,listener,new ByteArrayDeserializer());
	}
	public static void unRegister(String topic) {
		if (!topicMap.containsKey(topic)) {
			synchronized(topicMap){
				if (topicMap.containsKey(topic)) {
					ShuxKafkaConsumer<?> consumer = topicMap.get(topic);
					consumer.close();
					topicMap.remove(topic);
				}
			}
		}
		
		
	}
}
